package org.example;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.Message;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.MessageListener;
import org.apache.activemq.ActiveMQConnectionFactory;

import java.util.ArrayList;

class MyListenerOD implements MessageListener {

    int num = 0;
    double sum = 0;
    ArrayList arrayList = new ArrayList();

    public double average() {
        return sum/num;
    }

    public double variance() {
        double sum_var = 0;
        for(int i = 0; i < num; i++) {
            double d = (double)arrayList.get(i);
            sum_var += (d - sum/num) * (d - sum/num);
        }
        return Math.sqrt(sum_var/num);
    }

    public void onMessage(Message message) {
        try {
            String s = ((TextMessage)message).getText();
            double signal = Double.valueOf(s);
            num++;
            sum += signal;
            arrayList.add(signal);
            if(signal < average() - 3 * variance() || signal > average() + 3 * variance()) {
                System.out.println("第" + num + "个点是异常点：" + signal);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

public class OutlierDetection {
    public static void main(String[] args) throws JMSException {
        String brokerURL = "tcp://localhost:61616";
        ConnectionFactory factory = null;
        Connection connection = null;
        Session session = null;
        Topic topic = null;
        MessageConsumer messageConsumer = null;
        MyListenerOD listener = null;


        try {
            factory = new ActiveMQConnectionFactory(brokerURL);
            connection = factory.createConnection();

            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            topic = session.createTopic("MYTOPIC");

            messageConsumer = session.createConsumer(topic);

            listener = new MyListenerOD();

            messageConsumer.setMessageListener(listener);

            connection.start();

            System.out.println("Press any key to exit.");
            System.in.read();   // Pause
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            connection.close();
        }
    }
}
